package com.feiwo.nonego.rpc.proxy.api.future;

import com.feiwo.nonego.rpc.common.callback.AsyncRpcCallback;
import com.feiwo.nonego.rpc.common.threadpool.ClientThreadPool;
import com.feiwo.nonego.rpc.protocol.RpcProtocol;
import com.feiwo.nonego.rpc.protocol.enums.RpcStatus;
import com.feiwo.nonego.rpc.protocol.request.RpcRequest;
import com.feiwo.nonego.rpc.protocol.response.RpcResponse;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.locks.ReentrantLock;

/**
 *
 *
 * @author feiwo 2025-08-08
 */
public class RpcFuture extends CompletableFuture<Object> {

    private static final Logger LOGGER = LoggerFactory.getLogger(RpcFuture.class);

    private final RpcSync sync;
    private final RpcProtocol<RpcRequest> requestRpcProtocol;
    private RpcProtocol<RpcResponse> responseRpcProtocol;

    // 请求开始时间
    private final long startTime;
    // 响应时间阈值
    private static final long RESPONSE_TIME_THRESHOLD = 5000;

    private final List<AsyncRpcCallback> pendingCallbacks = new ArrayList<>();
    private final ReentrantLock lock = new ReentrantLock();

    public RpcFuture(RpcProtocol<RpcRequest> requestRpcProtocol) {
        this.sync = new RpcSync();
        this.requestRpcProtocol = requestRpcProtocol;
        this.startTime = System.currentTimeMillis();
    }

    @Override
    public boolean isDone() {
        return sync.isDone();
    }

    @Override
    public Object get() throws InterruptedException, ExecutionException {
        sync.acquire(-1);
        if (Objects.nonNull(responseRpcProtocol)) {
            return this.responseRpcProtocol.getBody().getResult();
        }else {
            return null;
        }
    }

    @Override
    public Object get(long timeout, TimeUnit unit) throws InterruptedException, ExecutionException, TimeoutException {
        var success = sync.tryAcquireSharedNanos(-1, unit.toNanos(timeout));
        if (success) {
            if (Objects.nonNull(responseRpcProtocol)) {
                return this.responseRpcProtocol.getBody().getResult();
            }else {
                return null;
            }
        }else {
            throw new TimeoutException("RpcFuture get timeout. requestId:"+
                    this.requestRpcProtocol.getHeader().getMsgId()+
                    ". Request class name: "+ this.requestRpcProtocol.getBody().getClassName()+
                    ". Request method name: "+ this.requestRpcProtocol.getBody().getMethodName());
        }
    }

    @Override
    public boolean isCancelled() {
        throw new UnsupportedOperationException();
    }

    @Override
    public boolean cancel(boolean mayInterruptIfRunning) {
        throw new UnsupportedOperationException();
    }

    public void done(RpcProtocol<RpcResponse> responseRpcProtocol) {
        this.responseRpcProtocol = responseRpcProtocol;
        sync.release(1);

        invokeCallbacks();

        var responseTime = System.currentTimeMillis() - startTime;
        LOGGER.debug("Service Response time Request id: {} Response Time: {} ms", responseRpcProtocol.getHeader().getMsgId(), responseTime);
        if (responseTime > RESPONSE_TIME_THRESHOLD) {
            LOGGER.warn("Service Response time is too slow. Request id: {}. Response Time: {} ms", responseRpcProtocol.getHeader().getMsgId(), responseTime);
        }
    }

    private void runCallback(final AsyncRpcCallback callback) {
        final var res = this.responseRpcProtocol.getBody();
        var header = this.responseRpcProtocol.getHeader();
        ClientThreadPool.submit(() -> {
            if (RpcStatus.SUCCESS.getCode() == header.getStatus()) {
                callback.onSuccess(res.getResult());
            }else {
                callback.onException(new RuntimeException("response error: "+ res.getError()));
            }
        });
    }

    public RpcFuture addCallback(AsyncRpcCallback callback) {
        this.lock.lock();
        try {
            if (isDone()) {
                runCallback(callback);
            }else {
                this.pendingCallbacks.add(callback);
            }
        }finally {
            this.lock.unlock();
        }
        return this;
    }

    private void invokeCallbacks() {
        this.lock.lock();
        try {
            for (AsyncRpcCallback callback : pendingCallbacks) {
                runCallback(callback);
            }
        }finally {
            this.lock.unlock();
        }
    }
}
